package cn.wygandwdn.core.transport.client;

import cn.wygandwdn.core.codec.MessageCodec;
import cn.wygandwdn.core.codec.ProtocolFrameDecoder;
import cn.wygandwdn.core.transport.client.handler.HeartBeatClientHandler;
import cn.wygandwdn.core.transport.client.handler.RpcClientPongHandler;
import cn.wygandwdn.core.transport.client.handler.RpcClientResponseHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @description:    因为可能有多个服务端,故需要存储客户端与服务端的连接通道-channel
 * @author: WYG
 * @time: 2021/10/27 16:05
 */

public class RpcChannel {
    private static final Logger log = LoggerFactory.getLogger(RpcChannel.class);

    private static Map<String, Channel> channels = new ConcurrentHashMap<>();

    private static EventLoopGroup worker;
    private static Bootstrap client = initialClient();

    public static Channel get(InetSocketAddress address) {
        String key = address.toString();
        // 当channel存在并处于激活状态时,直接返回channel即可
        if (channels.containsKey(key)) {
            Channel channel = channels.get(key);
            if (channel != null && channel.isActive()) {
                return channel;
            } else {
                channels.remove(key);
            }
        }
        Channel channel = null;

        try {
            channel = connect(client, address);
        } catch (ExecutionException | InterruptedException e) {
            log.error("客户端连接服务端发生错误: {}", e);
            return null;
        }

        channels.put(key, channel);
        return channel;
    }

    private static Channel connect(Bootstrap bootstrap, InetSocketAddress address) throws ExecutionException, InterruptedException {
        CompletableFuture<Channel> completableFuture = new CompletableFuture<>();
        ChannelFuture channelFuture = bootstrap.connect(address);
        channelFuture.addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                log.info("客户端连接服务端成功!");
                completableFuture.complete(future.channel());
            } else {
                throw new IllegalStateException();
            }
        });
        Channel channel = completableFuture.get();
        // 添加channel关闭时的回调接口,关闭EventLoopGroup,避免造成资源浪费
        channel.closeFuture().addListener(future -> {
            worker.shutdownGracefully();
        });
        return channel;
    }


    private static Bootstrap initialClient() {
        worker = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        MessageCodec CODEC = new MessageCodec();
        HeartBeatClientHandler HEART = new HeartBeatClientHandler();
        RpcClientPongHandler PONG = new RpcClientPongHandler();
        RpcClientResponseHandler RESPONSE = new RpcClientResponseHandler();

        bootstrap.group(worker)
                .channel(NioSocketChannel.class)
                // 连接超时时间,超过这个时间连接还未建立就代表连接失败
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                // 该参数会开启TCP对连接的监控,检测服务器是否处于活动状态,详情可以了解TCP保活机制:https://blog.springlearn.cn/posts/23666/
                .option(ChannelOption.SO_KEEPALIVE, true)
                // TCP默认开启Nagle算法,该算法的作用是尽可能发送大的数据块,不浪费网络带宽
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        // 添加LengthFieldBasedFrameDecoder,使后续的编解码器收到完整的报文
                        pipeline.addLast(new ProtocolFrameDecoder());
                        // 添加编自定义的解码器
                        pipeline.addLast(CODEC);
                        // 添加空闲检测事件处理器
                        pipeline.addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));
                        pipeline.addLast(HEART);
                        pipeline.addLast(PONG);
                        pipeline.addLast(RESPONSE);
                    }
                });
        return bootstrap;
    }
}
